package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TradeOrderBean;
import com.atguigu.util.ClickHouseUtil;
import com.atguigu.util.DateFormatUtil;
import com.atguigu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTradeOrderWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        String topicName = "dwd_course_order_detail";
        String groupId = "dws_trade_order_window";
        DataStreamSource<String> orderStream = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topicName, groupId));
        SingleOutputStreamOperator<JSONObject> mapStream = orderStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String provinceId = jsonObject.getString("province_id");
                if (provinceId != null) {
                    out.collect(jsonObject);
                }
            }
        });
        KeyedStream<JSONObject, String> keyedStream = mapStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2L))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts") * 1000;
                    }
                })).keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getString("user_id");
            }
        });
       // keyedStream.print();
        SingleOutputStreamOperator<TradeOrderBean> flatMapStream = keyedStream.flatMap(new RichFlatMapFunction<JSONObject, TradeOrderBean>() {
            ValueState<String> lastOrderDtState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> lastOrderDtStateDesc = new ValueStateDescriptor<>("last_order_dt", String.class);
                lastOrderDtStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L)).build());
                lastOrderDtState = getRuntimeContext().getState(lastOrderDtStateDesc);
            }

            @Override
            public void flatMap(JSONObject value, Collector<TradeOrderBean> out) throws Exception {
                String lastOrderDt = lastOrderDtState.value();
                Long ts = value.getLong("ts") * 1000;
                String toDate = DateFormatUtil.toDate(ts);
                // 下单人数
                Long orderUserCount = 0L;
                // 下单次数
                Long orderCount = 0L;
                if (lastOrderDt == null) {
                    orderUserCount += 1;
                    orderCount += 1;
                    lastOrderDtState.update(toDate);
                } else if (lastOrderDt.equals(toDate)) {
                    orderCount += 1;
                    lastOrderDtState.update(toDate);
                }
                out.collect(TradeOrderBean.builder()
                        .totalAmount(value.getDouble("final_amount"))
                        .orderUserCount(orderUserCount)
                        .orderCount(orderCount)
                        .ts(ts)
                        .build());

            }
        });
        SingleOutputStreamOperator<TradeOrderBean> reduceStream = flatMapStream.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)))
                .reduce(new ReduceFunction<TradeOrderBean>() {
                    @Override
                    public TradeOrderBean reduce(TradeOrderBean value1, TradeOrderBean value2) throws Exception {
                        value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                        value1.setOrderUserCount(value1.getOrderUserCount() + value2.getOrderUserCount());
                        value1.setTotalAmount(value1.getTotalAmount() + value2.getTotalAmount());
                        return value1;
                    }
                }, new AllWindowFunction<TradeOrderBean, TradeOrderBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<TradeOrderBean> values, Collector<TradeOrderBean> out) throws Exception {
                        TradeOrderBean next = values.iterator().next();
                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        next.setTs(System.currentTimeMillis());
                        out.collect(next);
                    }
                });
        reduceStream.addSink(ClickHouseUtil.getClickHouseSink("insert into dws_trade_order_window values(?,?,?,?,?,?)"));
        reduceStream.print();
        env.execute();
    }
}
